﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Threading;

namespace ReactiveTCPLibrary.Utilities
{
    public class ObservableAsyncReader<T> : IObserver<T>, IDisposable
    {
        class TaskToken
        {
            public bool IsEnd { get; set; }
            public Task<T> Task { get; set; }
            public Func<T> ValueGetter { get; set; }
        }

        volatile object _lock = new object();
        private Queue<TaskToken> tokens = new Queue<TaskToken>();
        private Queue<T> values = new Queue<T>();
        private int _working = 1;



        public Task<T> ReadAsync()
        {
            lock (_lock)
            {
                if (_working != 1)
                {
                    throw new ObjectDisposedException(this.GetType().Name, "This awaitable object is disposed, cannot provide any async value anymore.");
                }
                T value;

                if (values.Count > 0)
                {
                    value = values.Dequeue();
                    return Task.Factory.StartNew(() => value);
                }

                TaskToken t = new TaskToken();
                t.Task = new Task<T>(() => t.ValueGetter());
                tokens.Enqueue(t);
                return t.Task;
            }
        }


        public void OnCompleted()
        {
            Disposing(new IndexOutOfRangeException("The Observer you await is completed."));
        }

        public void OnError(Exception error)
        {
            Disposing(error);
        }

        private void Disposing(Exception error)
        {

            if (_working == 0)
                return;
            lock (_lock)
            {

                if (_working == 1)
                {
                    _working = 0;
                    TaskToken t;
                    while (tokens.Count > 0)
                    {
                        t = tokens.Dequeue();
                        t.ValueGetter = () => { throw new AggregateException(new[] { error }); };
                        t.Task.Start();
                    }
                    this.tokens = null;
                    this.values = null;
                }
            }

        }

        public void OnNext(T value)
        {
            TaskToken t;
            if (tokens.Count > 0)
            {
                t = tokens.Dequeue();
                t.ValueGetter = () => value;
                t.Task.Start();
                return;
            }
            values.Enqueue(value);
        }

        public void Dispose()
        {
            Disposing(new ObjectDisposedException(this.GetType().Name, "This awaitable object is disposed, cannot provide any async value anymore."));
        }
    }

}
